package com.tmt.im.wallet.service;

import com.tmt.im.common.enums.WalletEventType;
import com.tmt.im.common.pojo.*;
import com.tmt.im.wallet.helper.IgniteQueryHelper;
import com.tmt.springboot.common.annotation._ReflectInvoked;
import com.tmt.springboot.common.helper.IdSnowFlake;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.OffsetDateTime;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 定时服务
 *
 * @Description TODO
 * @Author wuyi
 * @Date 2024/9/4 13:28
 * @Version 1.0
 **/
@Slf4j
@Service
@Async("asyncTaskExecutor")
public class TimedTaskService {

    private final Ignite ignite;

    private final IgniteCache<RedPackageDetailKey, RedPackageDetailPoJo> redPackageDetailCache;

    private final IgniteCache<Long, WalletPoJo> walletCache;

    private final IgniteCache<Long, IMUserPoJo> userCache;

    private final IgniteCache<Long, WalletChangeDetailPoJo> walletDetailCache;

    private final IgniteCache<Long, RechargeHistoryPoJo> rechargeHistoryCache;

    private final RechargeHistoryService rechargeHistoryService;

    private final IgniteCache<Long, WithdrawHistoryPoJo> withdrawHistoryCache;

    private final WithdrawHistoryService withdrawHistoryService;

    public TimedTaskService(Ignite ignite,
                            RechargeHistoryService rechargeHistoryService,
                            WithdrawHistoryService withdrawHistoryService) {
        this.ignite = ignite;
        walletCache = ignite.cache(WalletPoJo.CACHE_NAME);
        walletDetailCache = ignite.cache(WalletChangeDetailPoJo.CACHE_NAME);
        redPackageDetailCache = ignite.cache(RedPackageDetailPoJo.CACHE_NAME);
        userCache = ignite.cache(IMUserPoJo.CACHE_NAME);
        rechargeHistoryCache = ignite.cache(RechargeHistoryPoJo.CACHE_NAME);
        withdrawHistoryCache = ignite.cache(WithdrawHistoryPoJo.CACHE_NAME);

        this.rechargeHistoryService = rechargeHistoryService;
        this.withdrawHistoryService = withdrawHistoryService;
    }

    @_ReflectInvoked("充值结果定时查询任务")
    @Scheduled(cron = "${timed-tasks.corn1}")
    public void timeQueryForRecharge() throws Exception {
        //找出超过一分钟未确认的充值历史记录列表。
        Timestamp expireTime = Timestamp.from((OffsetDateTime.now().minusMinutes(1)).toInstant());
        String sql = MessageFormat.format("select * from {0} where confirmMode = 0 and createTime < ?", RechargeHistoryPoJo.CACHE_NAME);
        List<RechargeHistoryPoJo> historyPoJos = IgniteQueryHelper.findAllBy(sql, new Object[]{expireTime},
                RechargeHistoryPoJo.class, rechargeHistoryCache);
        StringBuilder error = new StringBuilder();
        for (RechargeHistoryPoJo historyPoJo : historyPoJos) {
            error.setLength(0);
            if (rechargeHistoryService.queryRechargeResult(historyPoJo.getId(), error)) {
                rechargeHistoryService.updateWalletBalance(historyPoJo.getId());
            }
        }
    }

    @_ReflectInvoked("提现结果定时查询任务")
    @Scheduled(cron = "${timed-tasks.corn2}")
    public void timeQueryForWithdraw() throws Exception {
        //找出超过一分钟未确认的充值历史记录列表。
        Timestamp expireTime = Timestamp.from((OffsetDateTime.now().minusMinutes(1)).toInstant());
        String sql = MessageFormat.format("select * from {0} where confirmMode = 0 and createTime < ?", WithdrawHistoryPoJo.CACHE_NAME);
        List<WithdrawHistoryPoJo> historyPoJos = IgniteQueryHelper.findAllBy(sql, new Object[]{expireTime},
                WithdrawHistoryPoJo.class, withdrawHistoryCache);
        StringBuilder error = new StringBuilder();
        for (WithdrawHistoryPoJo historyPoJo : historyPoJos) {
            error.setLength(0);
            if (withdrawHistoryService.queryWithdrawResult(historyPoJo.getId(), error)) {
                withdrawHistoryService.updateWalletBalance(historyPoJo.getId());
            }
        }
    }

    @_ReflectInvoked("红包过期后退款的定时任务")
    @Scheduled(cron = "${timed-tasks.corn}")
    public void timedRefundForRedPackage() throws Exception {
        Timestamp timestamp = new Timestamp(System.currentTimeMillis());
        //查询所有红包详情记录中未领取的过期记录。
        String sql = MessageFormat.format("select * from {0} where receivedAlready = ? and refunded = ? and expireTime < ?",
                RedPackageDetailPoJo.CACHE_NAME);
        List<RedPackageDetailPoJo> list = IgniteQueryHelper.findAllBy(sql, new Object[]{false, false, timestamp},
                RedPackageDetailPoJo.class, redPackageDetailCache);
        if (CollectionUtils.isEmpty(list)) {
            return;
        }

        log.info("--------红包过期后退款的定时任务-----开始-------");
        try (IgniteSemaphore semaphore = ignite.semaphore("red_package_refund", 1, true, true)) {
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    //再次查询所有红包详情记录中未领取的过期记录。
                    list = IgniteQueryHelper.findAllBy(sql, new Object[]{false, false, timestamp},
                            RedPackageDetailPoJo.class, redPackageDetailCache);
                    if (!CollectionUtils.isEmpty(list)) {
                        //获取创建者列表
                        List<Long> creatorIds = list.stream().map(RedPackageDetailPoJo::getCreatorId).distinct().toList();
                        for (Long creatorId : creatorIds) {
                            List<RedPackageDetailPoJo> tmpList = list.stream()
                                    .filter(x -> Objects.equals(x.getCreatorId(), creatorId)).toList();
                            _refund(creatorId, tmpList);
                        }
                    }
                } catch (Exception ex) {
                    log.error(ex.getMessage(), ex);
                }
                log.info("--------红包过期后退款的定时任务-----结束-------");
            } else {
                log.info("--------红包过期后退款的定时任务-----结束(任务正在执行)-------");
            }
        }
    }

    private void _refund(Long creatorId, List<RedPackageDetailPoJo> list) {
        try (IgniteSemaphore semaphore = ignite.semaphore("wallet:" + creatorId, 1, true, true)) {
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    IMUserPoJo userPoJo = userCache.get(creatorId);
                    WalletPoJo walletPoJo = walletCache.get(userPoJo.getWalletId());
                    Map<Long, WalletChangeDetailPoJo> map = new LinkedHashMap<>();
                    List<Long> redPackageIds = list.stream().map(RedPackageDetailPoJo::getRedPackageId).distinct().toList();
                    for (Long redPackageId : redPackageIds) {
                        //检查该红包是否已经退款过了
                        String sql = MessageFormat.format("select * from {0} where eventType = ? and eventId = ?",
                                WalletChangeDetailPoJo.CACHE_NAME);
                        WalletChangeDetailPoJo detailPoJo = IgniteQueryHelper.findOneBy(sql, new Object[]{WalletEventType.REFUND_RED_PACKET.name(), redPackageId},
                                WalletChangeDetailPoJo.class, walletDetailCache);
                        if (detailPoJo != null) {
                            //已经退了，则直接处理下一个红包。
                            continue;
                        }

                        //统计需要退款的总额。
                        Long amount = list.stream()
                                .filter(x -> Objects.equals(x.getRedPackageId(), redPackageId))
                                .map(RedPackageDetailPoJo::getAmount)
                                .reduce(Long::sum)
                                .orElse(0L);

                        Long balance = walletPoJo.getBalance() + amount;
                        walletPoJo.setBalance(balance);

                        //创建钱包余额变更记录。
                        WalletChangeDetailPoJo walletDetailPoJo = WalletChangeDetailPoJo.builder()
                                .id(IdSnowFlake.getId())
                                .eventId(redPackageId)
                                .eventType(WalletEventType.REFUND_RED_PACKET.name())
                                .amount(amount)
                                .walletId(walletPoJo.getId())
                                .balance(balance)
                                .createTime(new Timestamp(System.currentTimeMillis()))
                                .creatorId(creatorId)
                                .version(0L)
                                .insert(true)
                                .build();
                        map.put(walletDetailPoJo.getKey(), walletDetailPoJo);
                    }

                    Map<RedPackageDetailKey, RedPackageDetailPoJo> detailPoJoMap = new LinkedHashMap<>();
                    for (RedPackageDetailPoJo detailPoJo : list) {
                        detailPoJo.setRefunded(true);
                        detailPoJoMap.put(detailPoJo.getKey(), detailPoJo);
                    }
                    //创建乐观串行事务。
                    try (Transaction tx = ignite.transactions().txStart(TransactionConcurrency.OPTIMISTIC,
                            TransactionIsolation.SERIALIZABLE)) {
                        walletCache.put(walletPoJo.getKey(), walletPoJo);
                        walletDetailCache.putAll(map);
                        redPackageDetailCache.putAll(detailPoJoMap);
                        //提交事务
                        tx.commit();
                    }

                } catch (Exception ex) {
                    log.error("红包过期后退款的定时任务异常:" + ex.getMessage(), ex);
                }
            } else {
                log.error("获取全局信号量失败");
            }
        }
    }

}
